package fnic.main;

import org.apache.commons.configuration.ConfigurationException;
import org.apache.log4j.Logger;

import fnic.prehand.action.ESClientHttpServer;
import fnic.prehand.common.AgentConfig;
import fnic.prehand.esagent.activity_measure.ActiveRedisRiver;
import fnic.prehand.esagent.passitive_measure2.PassitiveRedisRiver;
import fnic.prehand.esagent.ap_measure_new2.APRedisRiver;

public class PreHandMain {
	private static final Logger log = Logger.getLogger(PreHandMain.class);

//	private static final AgentConfig config = AgentConfig.getInstance();
//	private static int thdnum = config.getRedisthdnum()<cpucore ? config.getRedisthdnum() : cpucore;
	
	//单线程，one by one loop
//	private static void singleItemLoopCollect(AgentConfig config){
//		String ctrlURL = "http://" + config.getCtrlDBIp() + ":" + config.getCtrlDBPort()+ "/es/taskindex";
//		String skURL = "http://" + config. + ":" + webPort+ "/es/validatesession?
//		String pbAddrURL="http://"+config.getPbAddrIp()+":"+config.getPbAddrPort()+"/business/query?method=queryLocation&probeId="; //+probeId;
//		
//		Thread bulkexec = new Thread(new ActiveRedisRiver());
//		bulkexec.start();
//
//	}
	
	//线程池, pipeline bulk
//	private static void mutliItemsCollect(){
//		 ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
//		for(long i=0; i<3; i++){
////			scheduler.scheduleAtFixedRate(new Thread(new ActiveRedisRiver()), 1, 30, TimeUnit.SECONDS);
//		}
//	}
//	
//	//线程池, pipeline bulk
//	private static void mutliPassiveCollect(){
//		ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(5);
//		for(long i=0; i<5; i++){
//			scheduler.scheduleAtFixedRate(new Thread(new PassitiveRedisRiver()), 400, 100, TimeUnit.MILLISECONDS);
//		}
//	}
//	
//	private static void passiveLoopCollect(){
//		while(!stopfg){
//			Thread bulkexec = new Thread(new PassitiveRedisRiver());
//			bulkexec.start();
//			try{
//				bulkexec.join();
//			}catch(InterruptedException e){
//				e.printStackTrace();
//			}
//			stopfg = true;
//			
//		}
//	}
	
	public static void main(String[] args) {
		// 加载配置
		try {
			AgentConfig.getInstance().load(); 
		} catch (ConfigurationException e) {
			log.error("load es_agent configuration failed: " + e.getMessage());
			return;
		}
		
		AgentConfig config = AgentConfig.getInstance();
		/*---------------------------主动拨测项目启动-------------------------*/
		if(config.isMujob_on()){
//			int mujob_httpport = config.getMujob_httpPort();
//			String mujob_logname = config.getMujob_logName();
			String mujob_index = config.getMujob_IndexName();
			String mujob_type = config.getMujob_IndexType();
			String mujob_redis_host = config.getMujob_redisHost();
			int mujob_redis_port = config.getMujob_redisPort();
			String mujob_redis_key = config.getMujob_redisKey();
			String mojub_ctrlURL = "http://" + config.getMujob_ctrlIp() + ":" + config.getMujob_ctrlPort();
			String mujob_skURL = "http://" + config.getMujob_skValidIp() + ":" + config.getMujob_skValidPort()+ "/es/validatesession?";
			String mujob_pbAddrURL="http://"+config.getMujob_pbAddrIp()+":"+config.getMujob_pbAddrPort()+"/business/query?method=queryLocation&probeId="; //+probeId;
			
//			BlockQueueService mujob_queue = new BlockQueueService(); 
//			Thread httpserver = new Thread(new ESClientHttpServer(mujob_httpport, mujob_logname, mujob_skURL, mujob_index, mujob_type, mujob_queue));
//			Thread queuepull = new Thread(new ActiveLocalQueueRiver(mujob_index, mujob_type, mojub_ctrlURL, mujob_skURL, mujob_queue));
			Thread bulkexec = new Thread(new ActiveRedisRiver(mujob_index, mujob_type, mujob_redis_host,mujob_redis_port, mujob_redis_key, mojub_ctrlURL, mujob_skURL, mujob_pbAddrURL));
//			httpserver.start();
//			queuepull.start();
			bulkexec.start();
		}

		
		/*---------------------------主动网络中心项目启动-------------------------*/
		if(config.isNetcenter_on()){
//			int nc_httpport = config.getNc_httpPort();
//			String nc_logname = config.getNc_logName();
			String nc_index = config.getNc_IndexName();
			String nc_type = config.getNc_IndexType();
			String nc_redis_host = config.getNc_redisHost();
			int nc_redis_port = config.getNc_redisPort();
			String nc_redis_key = config.getNc_redisKey();
			String nc_ctrlURL = "http://" + config.getNc_ctrlIp() + ":" + config.getNc_ctrlPort();
			String nc_skURL = "http://" + config.getNc_skValidIp() + ":" + config.getNc_skValidPort()+ "/es/validatesession?";
			String nc_pbAddrURL="http://"+config.getNc_pbAddrIp()+":"+config.getNc_pbAddrPort()+"/business/query?method=queryLocation&probeId="; //+probeId;
			
//			BlockQueueService nc_queue = new BlockQueueService(); 
//			Thread nc_httpserver = new Thread(new ESClientHttpServer(nc_httpport, nc_logname, nc_skURL, nc_index, nc_type, nc_queue));
//			Thread nc_queuepull = new Thread(new ActiveLocalQueueRiver(nc_index, nc_type, nc_ctrlURL, nc_skURL, nc_queue));
			Thread nc_bulkexec = new Thread(new ActiveRedisRiver(nc_index, nc_type, nc_redis_host,nc_redis_port, nc_redis_key,nc_ctrlURL, nc_skURL, nc_pbAddrURL));
//			nc_httpserver.start();
//			nc_queuepull.start();
			nc_bulkexec.start();
			
		}
		
		
		/*---------------------------被动项目启动-------------------------*/
		if(config.isPassitive_on()){
			String p_redis_host = config.getPredisHost();
			int p_redis_port = config.getPredisPort();
			String p_redis_key = config.getPredisKey();
			Thread pass_index = new Thread(new PassitiveRedisRiver(p_redis_host, p_redis_port, p_redis_key));
			pass_index.start();
			
//			try {
//				pass_index.join();
//			} catch (InterruptedException e) {
//				// TODO Auto-generated catch block
//				e.printStackTrace();
//			}
		}
		
		
		/*---------------------------AP数据采集-------------------------*/
		if(config.isAp_on()){
			String ap_redis_host = config.getApredisHost();
			int ap_redis_port = config.getApredisPort();
			String ap_redis_key = "staflow";
//			BlockQueueService ap_queue = new BlockQueueService(); 
			Thread ap_index = new Thread(new APRedisRiver(ap_redis_host, ap_redis_port, ap_redis_key));
			ap_index.start();
			
//			try {
//				ap_index.join();
//			} catch (InterruptedException e) {
//				// TODO Auto-generated catch block
//				e.printStackTrace();
//			}
		}
		
		if(config.isMujob_on()||config.isNetcenter_on()||config.isAp_on()){
			Thread httpserver = new Thread(new ESClientHttpServer());
			httpserver.start();
			try {
				httpserver.join();
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
//		mutliPassiveCollect();
	}

}
